package com.shujia.streaming

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo2UpdateStateByKey {
  def main(args: Array[String]): Unit = {
    /**
     * spark core: SparkContext  --> 核心数据类型 RDD
     * spark sql: SparkSession --> 核心数据类型 DataFrame
     * spark streaming: StreamingContext --> 核心数据类型 DStream(RDD)
     *
     * 创建SparkStreaming的环境需要使用StreamingContext对象
     */
    //spark streaming依赖于spark core的环境
    //因为spark streaming核心数据类型底层封装的是RDD
    val conf = new SparkConf()
    conf.setMaster("local[2]")
    conf.setAppName("spark streaming 单词统计案例1")
    val sparkContext = new SparkContext(conf)
    // def this(sparkContext: SparkContext, batchDuration: Duration) //传入一个spark core 环境对象以及一个接收时间的范围
    val streamingContext = new StreamingContext(sparkContext, Durations.seconds(5))
    //设置checkpoint路径
    streamingContext.checkpoint("spark/data/stream_state_wc_checkpoint")

    /**
     * 因为sparkstreaming是一个近实时的计算引擎，整个程序写完运行的状态，一直运行，一直接收数据，除非异常或者手动停止
     *
     * 1、目前在spark使用nc工具监控一个端口号中源源不断产生的数据
     * 2、yum install -y nc 下载安装nc工具
     * 3、nc -lk xxxx
     */
    val lineDS: ReceiverInputDStream[String] = streamingContext.socketTextStream("master", 12345)

    val flatMapDS: DStream[String] = lineDS.flatMap(_.split(" "))
    val wordsKVDS: DStream[(String, Int)] = flatMapDS.map((_, 1))

    /**
     * 有状态的算子
     * 使用当前批次的数据加上之前批次统计的结果，返回新的结果
     *
     * seq: 代表当前批次一个key对应的所有value值  (hello,1)  seq:[1,1,1]
     * Option： 之前批次统计的key对应的值，如果当前批次的key在历史批次中没有出现，那么Option就是None (hello,2) Option:2
     *
     *
     * 有状态算子需要配合checkpoint一起使用，将上一次的结果持久化，为了下一批次可以读取到
     */
    //def updateStateByKey[S: ClassTag](
    //      updateFunc: (Seq[V], Option[S]) => Option[S]
    //    ): DStream[(K, S)]
    val result2DS: DStream[(String, Int)] = wordsKVDS.updateStateByKey((seq: Seq[Int], option: Option[Int]) => {
      //将当前批次该key对应的所有value值加起来
      val currentCount: Int = seq.sum // [1,1,1]->3
      //获取上一批次的该key对应的value值
      val count: Int = option.getOrElse(0) // 2
      //将当前匹配的key对应的value值总和与上一次该key的value值相加，封装成一个Option
      Option(currentCount + count)
    })

    //打印一下结果
    result2DS.print()

    /**
     * Spark Streaming作业触发执行的方式和之前完全不一样
     */
    streamingContext.start()
    streamingContext.awaitTermination()
    streamingContext.stop()


  }
}
